[WIP] Add semaphore to wait for streams to become available. #38
Conversation
| else: | ||
| self.state = ConnectionState.ACTIVE | ||
| finally: | ||
| self.streams_semaphore.release() |
There was a problem hiding this comment.
Okay, so I think the wrapping here isn't quite correct. It'd make sense to instead wrap up the portion of this method after the connection init, so...
# Wrap this up inside the stream semaphore
h2_stream = AsyncHTTP2Stream(stream_id=stream_id, connection=self)
self.streams[stream_id] = h2_stream
self.events[stream_id] = []
return await h2_stream.request(method, url, headers, stream, timeout)Since that'd ensure that we hold the semaphore for the duration of a request/response.
In fact, it's actually a bit more awkward that tho. We don't want to release the semaphore until the client has closed the response we've returned. So really we want to release the semaphore either on an exception occuring within this block, or on the response being closed.
There was a problem hiding this comment.
So, something like...
async def request(...):
...
await self.streams_semaphore.acquire()
try:
h2_stream = AsyncHTTP2Stream(stream_id=stream_id, connection=self)
self.streams[stream_id] = h2_stream
self.events[stream_id] = []
return await h2_stream.request(method, url, headers, stream, timeout)
except:
self.streams_semaphore.release()
raise
...
async def close_stream(self, stream_id: int) -> None:
self.streams_semaphore.release()
...Alternately we could pass the semaphore to AsyncHTTP2Stream, and wrap up the behaviour there instead. (That approach might feel a bit more tightly constrained)
|
Also, answers to some of the questions here...
|
|
Closing this now in favour of #89 - thanks so much for your initial time on it - it's been really helpful! |
This PR adds a semaphore lock to prevent exceptions when the
MAX_CONCURRENT_STREAMSlimit is reached. It now instead waits for a stream to close instead of raising an exception.This bug was initially reported in httpx.
Some questions / comments:
What timeout should be used (if any)?
Both
AsyncSemaphoreandSyncSemaphoreallow a timeout to be passed when acquiring. Should my code accept any sort of timeout?Should semaphores support
withstatements?The abstract semaphore interface doesn't implement any entry or exit functions, preventing the use of the
withstatement. I presume this is because this would ignore the timeout condition, and could lead to programming errors in the future.Here are some possible ideas:
What exception should be thrown if a semaphore timeout is reached?
I'm currently using
PoolTimeout, but it's a noop becuase without a timeout it's never going to be called anyway. Should I just passNone?